package com.education


import org.apache.spark._
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.spark.streaming.dstream.DStream

/**
 * Hello world!
 *
 */
  object SparkStreaming_poi86{
  def main(args: Array[String]): Unit = {
    val sparkconf = new SparkConf().setAppName("poi").setMaster("spark://192.168.157.100:7077")
    val sc = new SparkContext(sparkconf)
    val ssc = new StreamingContext(sc,Seconds(10))
    val topicLines = KafkaUtils.createStream(ssc,"192.168.157.100:2181","poi20180531",Map("poi86" -> 1))
    val lines = topicLines.map(_._2.replace("\"",""))
    val props = new Properties()
    props.put("bootstrap.servers", "\"192.168.157.100:9092,\"192.168.157.101:9092,\"192.168.157.102:9092")
    props.put("key.serializer", classOf[StringSerializer].getName)
    props.put("value.serializer", classOf[StringSerializer].getName)

    process(lines,props)
    ssc.start()
    ssc.awaitTermination()
  }
  def process(lines: DStream[String],props:java.util.Properties)={
    lines.foreachRDD(
      rdd=>rdd.foreachPartition(partition=> {
        val producer = new KafkaProducer[String, String](props)
        partition.foreach(
          record => {
            val line = record.toString
            if (line.split(",").length >6 && !line.split(",")(6).isEmpty) {
              producer.send(new ProducerRecord[String, String]("test", line))
            }
          }
        )
        producer.close();
      }
    ))
  }
}

